package com.zhss.dfs.upgrade;

import com.zhss.dfs.server.DataNodeConfig;
import com.zhss.dfs.server.NameNodeServiceActor;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.CacheRequest;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @program: filesystem
 * @description:
 * @author: Liu Weidong
 * @create: 2024-05-14 21:58
 **/
public class NioServer extends Thread {

    public static final Integer PROCESSOR_NUM = 1;

    public static final Integer IO_THREAD_NUM = 1;

    private Selector selector;

    public static void main(String[] args) {
        NioServer nioServer = new NioServer(null);
        nioServer.start();
    }

    private List<Processor> processors = new ArrayList<>();

    public NioServer(NameNodeServiceActor nameNodeServiceActor) {
        ServerSocketChannel serverSocketChannel = null;
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.bind(new InetSocketAddress(DataNodeConfig.DATANODE_HOSTNAME, DataNodeConfig.DATANODE_PORT));
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            NetworkResponseQueue networkResponseQueue = NetworkResponseQueue.get();
            for (int i = 0; i < PROCESSOR_NUM; i++) {
                // 开启多个processor线程
                String processorId = UUID.randomUUID().toString();
                networkResponseQueue.initResponseQueue(processorId);
                Processor processor = new Processor(processorId);
                processors.add(processor);
                processor.start();
            }

            for (int i = 0; i < IO_THREAD_NUM; i++) {
                new IOThread(nameNodeServiceActor).start();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        try {
            System.out.println("dataNode-fileServer已经启动监听端口号" + DataNodeConfig.DATANODE_PORT + "...........................");
            while (true) {
                selector.select();
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    if (key.isAcceptable()) {
                        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        if (socketChannel != null) {
                            socketChannel.configureBlocking(false);
                            int processorIndex = new Random().nextInt(PROCESSOR_NUM);
                            Processor processor = processors.get(processorIndex);
                            // 交给processor进行处理
                            processor.addChannel(socketChannel);
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
